package t20240514_lineage;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.*;


public class Test3 {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);

        String sql ="create table o_scdn_vod_xserver_ngx_httpcode (\n" +
                "  `stime` string,\n" +
                "  `type`  string,\n" +
                "  `hostname`  string,\n" +
                "  `domain`  string,\n" +
                "  `status`  int,\n" +
                "  `status_cnt`  int,\n" +
                "  `port`  int, \n" +
                "  `level` string,\n" +
                "  rowtime as to_timestamp(`stime`),\n" +
                "  proctime as proctime(),\n" +
                "  watermark for rowtime as rowtime - interval '20' second\n" +
                ") with (\n" +
                "  'properties.bootstrap.servers' = 'bdyun-suzhou-tw04a0015.su.baidu.internal:6667,bdyun-suzhou-tw04b0023.su.baidu.internal:6667,bdyun-suzhou-tw04b0025.su.baidu.internal:6667,bdyun-suzhou-tw04d0583.su.baidu.internal:6667,bdyun-suzhou-tw04d0581.su.baidu.internal:6667',\n" +
                "  'connector' = 'kafka',\n" +
                "  'format' = 'avro-event',\n" +
                "  'topic' = 'o_scdn_vod_xserver_ngx_httpcode',\n" +
                "  'properties.group.id' = 'flink-lichen-o_scdn_vod_xserver_ngx_httpcode-lc-kafka-scdn2',\n" +
                "  'properties.security.protocol' = 'SASL_PLAINTEXT',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'properties.sasl.kerberos.service.name' = 'kafka'\n" +
                ");\n" +
                "create table t_domain_of_data_platform (\n" +
                "  `userid`  bigint,\n" +
                "  `domain_name` string\n" +
                ") with (\n" +
                "  'connector' = 'tidb',\n" +
                "  'driver' = 'com.mysql.jdbc.Driver',\n" +
                "  'lookup.cache.ttl' = '30min',\n" +
                "  'table-name' = 't_domain_of_data_platform',\n" +
                "  'url' = 'jdbc:mysql://tidb-lb.data.onething.internal:4000/dim?useUnicode=true&useSSL=false&characterEncoding=UTF-8&allowMultiQueries=true&autoReconnect=true',\n" +
                "  'lookup.max-retries' = '3',\n" +
                "  'username' = 'beta',\n" +
                "    'password' = 'beta@9898w',\n" +
                "  'query-sql' = 'select account_id as userid, domain_name from t_domain_of_data_platform'\n" +
                ");\n" +
                "create table app_i1_scdn_vod_xserver_httpcode_stat (\n" +
                "  `stime` timestamp,\n" +
                "  `grpid` int,\n" +
                "  `account_id`  string,\n" +
                "  `domain`  string,\n" +
                "  `hostname`  string,\n" +
                "  `level` string,\n" +
                "  `status`  int,\n" +
                "  `status_cnt_total`  int\n" +
                ") with (\n" +
                "  'database' = 'scdn_app_lo',\n" +
                "  'password' = 'rNXB>=dQCe3Ri-CGg*B?',\n" +
                "  'table-mode' = 'single',\n" +
                "  'sink.write-mode' = 'local',\n" +
                "  'connector' = 'clickhouse',\n" +
                "  'sink.flush-interval' = '10s',\n" +
                "  'sink.batch-size' = '10000',\n" +
                "  'url' = 'clickhouse://ck-lb.data.onething.internal:8123',\n" +
                "  'table' = 'app_i1_scdn_vod_xserver_httpcode_stat',\n" +
                "  'username' = 'spark'\n" +
                ");\n" +
                "\n" +
                "create view o_scdn_vod_xserver_ngx_httpcode_fliter as\n" +
                "select\n" +
                "    *\n" +
                "from o_scdn_vod_xserver_ngx_httpcode\n" +
                "where\n" +
                "    (\n" +
                "    (level = 'edge' and `port` in (80, 443, 9102, 9305))\n" +
                "    or (level = 'area' and `port` in (80, 9102))\n" +
                "    or (level = 'root' and `port` = 80)\n" +
                "    )\n" +
                "    and `domain` <> '\\N'\n" +
                "    and `hostname` <> '\\N'\n" +
                "    and level <> '\\N';\n" +
                "\n" +
                "create view o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules as\n" +
                "select\n" +
                "    ori.`hostname`,\n" +
                "    ori.`domain` as field,\n" +
                "    ori.`domain`,\n" +
                "    ori.`status`,\n" +
                "    ori.`level`,\n" +
                "    ori.`status_cnt`,\n" +
                "    ori.`rowtime`,\n" +
                "    proctime() as proctime\n" +
                "from o_scdn_vod_xserver_ngx_httpcode_fliter ori;\n" +
                "\n" +
                "create view o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules_plat as\n" +
                "select\n" +
                "    ifnull(split_index(oria.`field`, '|', 1), cast(plat.userid as string)) `account_id`,\n" +
                "    oria.`hostname`,\n" +
                "    split_index(oria.`field`, '|', 0) as dom,\n" +
                "    oria.`status`,\n" +
                "    oria.`level`,\n" +
                "    oria.`status_cnt`,\n" +
                "    oria.`rowtime`\n" +
                "from o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules oria\n" +
                "left join t_domain_of_data_platform for system_time as of oria.proctime plat on split_index(oria.`field`, '|', 0) = plat.`domain_name`;\n" +
                "\n" +
                "create view o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules_plat_group as\n" +
                "select\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    cast(`account_id` as string) `account_id`,\n" +
                "    `hostname`,\n" +
                "    `dom`,\n" +
                "    `status`,\n" +
                "    `level`,\n" +
                "    sum(`status_cnt`) as `status_sum`\n" +
                "from table(tumble(table o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules_plat, descriptor(`rowtime`), interval '60' second))\n" +
                "group by `account_id` ,`hostname`, `dom`, `status`, `level`, window_start, window_end;\n" +
                "\n" +
                "insert into app_i1_scdn_vod_xserver_httpcode_stat\n" +
                "select\n" +
                "    window_start as stime,\n" +
                "    1 `grpid`,\n" +
                "    case `account_id` when 'null' then null else `account_id` end as `account_id`,\n" +
                "    `dom` as `domain`,\n" +
                "    '-9999' as `hostname`,\n" +
                "    `level`,\n" +
                "    `status`,\n" +
                "    sum(`status_sum`) as `status_cnt_total`\n" +
                "from o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules_plat_group\n" +
                "group by  case `account_id` when 'null' then null else `account_id` end , `dom`, `status`, `level`, window_start, window_end\n" +
                "union all\n" +
                "select\n" +
                "    window_start as stime,\n" +
                "    2 `grpid`,\n" +
                "    case `account_id` when 'null' then null else `account_id` end as `account_id`,\n" +
                "    `dom` as `domain`,\n" +
                "    `hostname`,\n" +
                "    `level`,\n" +
                "    `status`,\n" +
                "    sum(`status_sum`) as `status_cnt_total`\n" +
                "from o_scdn_vod_xserver_ngx_httpcode_fliter_join_rules_plat_group\n" +
                "group by case `account_id` when 'null' then null else `account_id` end , `dom`,`hostname`,`level`,`status`, window_start, window_end;";


        String[] statements = getStatements(sql);

        LineageContext lineageContext = new LineageContext(tbEnv);
        Tuple2<List<LineageRel>, LineageResult> tuple2 = lineageContext.analyzeLineage(statements);
        List<LineageRel> lineageRels = tuple2.f0;

        getTestJson(lineageRels);

        LineageResult f1 = tuple2.f1;

        JSONObject o = (JSONObject)JSONObject.toJSON(f1);
        System.out.println();
        System.out.println(o.toJSONString());


    }


    private static void getTestJson(List<LineageRel> lineageRels) {


        Map<String, Set<String>> nodes = new HashMap<>();


        // edges
        JSONArray array = new JSONArray();
        for (LineageRel rel : lineageRels) {
            JSONObject from = new JSONObject();
            from.put("name", rel.getSourceTable());
            from.put("field", rel.getSourceColumn());

            JSONObject to = new JSONObject();
            to.put("name", rel.getTargetTable());
            to.put("field", rel.getTargetColumn());


            addNode(nodes, rel.getSourceTable(), rel.getSourceColumn());
            addNode(nodes, rel.getTargetTable(), rel.getTargetColumn());


            JSONObject object = new JSONObject();

            object.put("from", from);
            object.put("to", to);

            array.add(object);
        }

        // nodes

        JSONArray nodeArrays = new JSONArray();
        for (Map.Entry<String, Set<String>> stringSetEntry : nodes.entrySet()) {
            String tableName = stringSetEntry.getKey();
            JSONObject node = new JSONObject();
            node.put("name", tableName);

            JSONArray fileds = new JSONArray();
            for (String s : stringSetEntry.getValue()) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("name", s);
                fileds.add(jsonObject);
            }
            node.put("fields", fileds);

            nodeArrays.add(node);

        }


        JSONObject result = new JSONObject();
        result.put("edges", array);
        result.put("nodes", nodeArrays);

        System.out.println(result.toJSONString());
    }


    private static void addNode(Map<String, Set<String>> nodes,String tableName, String field) {
        Set<String> sourceFields = nodes.get(tableName);
        if(sourceFields != null) {
            sourceFields.add(field);
        } else {
            Set<String> list = new HashSet<>();
            list.add(field);
            nodes.put(tableName, list);
        }
    }
    public static String[] getStatements(String sql) {


        final String localSqlSeparator = ";\\s*(?:\\n|--.*)";
        String[] splits = sql.replace("\r\n", "\n").split(localSqlSeparator);
        String lastStatement = splits[splits.length - 1].trim();
        if (lastStatement.endsWith(";")) {
            splits[splits.length - 1] = lastStatement.substring(0, lastStatement.length() - 1);
        }

        return splits;
    }
}
